#pragma once

#ifndef __MY_MESSAGE__
#define __MY_MESSAGE__

#include "../mqcommon/message.pb.h"
#include "../mqcommon/helpper.hpp"

#include <iostream>
#include <string>
#include <cassert>
#include <memory>
#include <list>
#include <unordered_map>

namespace NS_Message
{
    using MessagePtr = std::shared_ptr<Message>;
    class MessagePersist
    {
        const std::string file_subfix = ".mqd";
        const std::string tmp_file_subfix = ".mqd.tmp";

    public:
        MessagePersist(const std::string &dirname, const std::string &queue_name)
            : _queue_name(queue_name)
        {
            std::string dir = dirname.back() == '/' ? dirname : dirname + "/";
            NS_Helper::FileHelper::createDirectory(dir);
            _filename = dir + _queue_name + file_subfix;
            _tmp_filename = dir + _queue_name + tmp_file_subfix;
        }

        void createFile()
        {
            assert(NS_Helper::FileHelper::createFile(_filename));
        }
        bool deleteFile()
        {
            return NS_Helper::FileHelper::deleteFile(_filename) && NS_Helper::FileHelper::deleteFile(_tmp_filename);
        }

        const int len_format = 5;

        // 向文件中插入新的消息，只添加消息的body部分
        bool insert(MessagePtr &ptr)
        {
            return insert(_filename, ptr);
        }

        const std::string valid = "1";
        const std::string invalid = "0";

        void erase(const MessagePtr &ptr)
        {
            NS_Helper::FileHelper file_helper(_filename);
            // 1. 将消息置为无效消息
            ptr->mutable_payload()->set_valid(invalid);
            // 2. 将消息序列化，获取body部分
            std::string body = ptr->payload().SerializeAsString();
            if (body.size() + len_format != ptr->len())
            {
                ELOG("消息长度与文件中的消息长度不一致，消息长度：%ld， 文件中消息长度：%d", body.size() + len_format, ptr->len());
                return;
            }
            // 3. 获取消息持久化的位置，根据该位置覆盖式写入新的消息

            std::string len;
            // DLOG("消息长度：%ld", body.size());
            NS_Helper::DecimalConversion::decimalToBase62(body.size(), len);
            std::string msg = len + body;
            file_helper.write(msg, ptr->offset(), msg.size());
        }

        // void erase(const MessagePtr &ptr)
        // {
        //     NS_Helper::FileHelper file_helper(_filename);
        //     // 1. 将消息置为无效消息
        //     ptr->mutable_payload()->set_valid(invalid);
        //     // 2. 将消息序列化，获取body部分
        //     std::string body = ptr->payload().SerializeAsString();
        //     if (body.size() + len_format != ptr->len())
        //     {
        //         ELOG("消息长度与文件中的消息长度不一致，消息长度：%ld， 文件中消息长度：%d", body.size() + 8, ptr->len());
        //         return;
        //     }
        //     // 3. 获取消息持久化的位置，根据该位置覆盖式写入新的消息

        //     std::stringstream ss;
        //     ss << std::setw(len_format) << std::setfill('0') << body.size();
        //     std::string len = ss.str();
        //     std::string msg = len + body;
        //     file_helper.write(msg, ptr->offset(), msg.size());
        // }
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> ret;
            // 读取队列文件中的有效消息到内存中
            if (!load(ret))
            {
                ELOG("读取队列文件失败");
                return ret;
            }
            DLOG("gc后有效队列大小：%ld", ret.size());

            // 写入有效消息到替换队列文件中
            NS_Helper::FileHelper::createFile(_tmp_filename);
            NS_Helper::FileHelper tmp_file_helper(_tmp_filename);
            for (auto &ptr : ret)
            {
                if (!insert(_tmp_filename, ptr))
                {
                    ELOG("写入消息到替换队列文件失败");
                    return ret;
                }
            }
            // 删除队列文件
            if (!NS_Helper::FileHelper::deleteFile(_filename))
            {
                ELOG("删除源队列文件失败");
                return ret;
            }
            // 重命名替换队列文件名为队列文件
            if (!NS_Helper::FileHelper::renameFile(_tmp_filename, _filename))
            {
                ELOG("重命名队列文件失败 %s -> %s ", _tmp_filename.c_str(), _filename.c_str());
                return ret;
            }
            return ret;
        }

        ~MessagePersist() {}

    private:
    // 向文件中写入消息数据
    bool insert(const std::string &filename, MessagePtr &ptr)
    {
        NS_Helper::FileHelper file_helper(filename);
        // 1. 获取文件的大小
        size_t filesize = file_helper.size();
        // 2. 获取消息的body部分的序列化
        std::string body = ptr->payload().SerializeAsString();
        // 3. 向文件中写入结构为长度（十字节） + 消息 为了之后的消息恢复做准备
        // std::stringstream ss;
        int len = body.size();
        std::string base62;
        if(!NS_Helper::DecimalConversion::decimalToBase62(len, base62))
        {
            DLOG("从十进制转换到62进制失败，%d", len);
            return false;
        }

        std::string msg = base62 + body;
        if (file_helper.write(msg, filesize, msg.size()) < 0)
        {
            return false;
        }
        // 4. 更新内存中消息的持久化的消息的位置和长度
        ptr->set_len(msg.size());
        ptr->set_offset(filesize);

        return true;
    }

    // bool insert(const std::string &filename, MessagePtr &ptr)
        // {
        //     NS_Helper::FileHelper file_helper(filename);
        //     // 1. 获取文件的大小
        //     size_t filesize = file_helper.size();
        //     // 2. 获取消息的body部分的序列化
        //     std::string body = ptr->payload().SerializeAsString();
        //     // 3. 向文件中写入结构为长度（十字节） + 消息 为了之后的消息恢复做准备
        //     std::stringstream ss;
        //     ss << std::setw(len_format) << std::setfill('0') << body.size();
        //     std::string len = ss.str();

        //     // std::cout << "len.size(): " << len.size() << "#" << len << std::endl;
        //     // std::cout << "body.size(): " << body.size() << "#" << body << std::endl;

        //     std::string msg = len + body;
        //     // std::cout << "msg.size(): " << msg.size() << "#" << msg << std::endl;
        //     if (file_helper.write(msg, filesize, msg.size()) < 0)
        //     {
        //         return false;
        //     }
        //     // 4. 更新内存中消息的持久化的消息的位置和长度
        //     ptr->set_len(msg.size());
        //     ptr->set_offset(filesize);

        //     return true;
        // }

        bool load(std::list<MessagePtr> &ret)
        {
            NS_Helper::FileHelper file_helper(_filename);
            size_t offset = 0, filesize = file_helper.size();
            while (offset < filesize)
            {
                // 读取五字节的62位消息长度
                std::string str_len;
                if (!file_helper.read(str_len, offset, len_format))
                {
                    ELOG("读取消息长度失败");
                    return false;
                }

                try
                {
                    size_t len = 0;
                    NS_Helper::DecimalConversion::base62ToDecimal(str_len, len);
                    offset += len_format;
                    // 根据长度读取完整的消息
                    std::string body;
                    if (!file_helper.read(body, offset, len))
                    {
                        ELOG("读取消息失败");
                        return false;
                    }
                    offset += len;

                    MessagePtr ptr = std::make_shared<Message>();
                    ptr->mutable_payload()->ParseFromString(body);
                    // 将消息反序列化为Message对象并添加到list中
                    // 如果消息无效，则重新获取消息
                    if (ptr->mutable_payload()->valid() == "0")
                        continue;
                    ret.push_back(ptr);
                }
                catch (const std::invalid_argument &e)
                {
                    ELOG("读取的消息长度不是有效的整数: %s", str_len.c_str());
                    return false;
                }
                catch (const std::out_of_range &e)
                {
                    ELOG("读取的消息长度超出范围: %s", str_len.c_str());
                    return false;
                }
            }

            return true;
        }

        // bool load(std::list<MessagePtr> &ret)
        // {
        //     NS_Helper::FileHelper file_helper(_filename);
        //     size_t offset = 0, filesize = file_helper.size();
        //     while (offset < filesize)
        //     {
        //         // 读取十字节整形的消息长度
        //         std::string str_len;
        //         if (!file_helper.read(str_len, offset, len_format))
        //         {
        //             ELOG("读取消息长度失败");
        //             return false;
        //         }

        //         try
        //         {
        //             size_t len = std::stoi(str_len);
        //             offset += len_format;
        //             // 根据长度读取完整的消息
        //             std::string body;
        //             if (!file_helper.read(body, offset, len))
        //             {
        //                 ELOG("读取消息失败");
        //                 return false;
        //             }
        //             offset += len;

        //             MessagePtr ptr = std::make_shared<Message>();
        //             ptr->mutable_payload()->ParseFromString(body);
        //             // 将消息反序列化为Message对象并添加到list中
        //             // 如果消息无效，则重新获取消息
        //             if (ptr->mutable_payload()->valid() == "0")
        //                 continue;
        //             ret.push_back(ptr);
        //         }
        //         catch (const std::invalid_argument &e)
        //         {
        //             ELOG("读取的消息长度不是有效的整数: %s", str_len.c_str());
        //             return false;
        //         }
        //         catch (const std::out_of_range &e)
        //         {
        //             ELOG("读取的消息长度超出范围: %s", str_len.c_str());
        //             return false;
        //         }
        //     }

        //     return true;
        // }

    private:
        std::string _queue_name;
        std::string _filename;
        std::string _tmp_filename;
    };

    class QueueMessage
    {
    public:
        using Ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(const std::string &basedir, const std::string &qname) : _qname(qname), _persist(basedir, qname)
        {
            _persist.createFile();
        }

        void recovery()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            // 恢复历史文件中消息数据
            _to_push_msgs = _persist.gc();
            for (auto &ptr : _to_push_msgs)
            {
                _durable_msgs[ptr->payload().property().id()] = ptr;
            }
            _total_count = _valid_count = _to_push_msgs.size();
            DLOG("%s恢复历史消息，消息数量：%d", _qname.c_str(), (int)_to_push_msgs.size());
        }

        // 第三个参数是当用户没有给该消息设定参数时，消息的持久化就会取决于归属队列的持久化状态
        // 接受消息到待推送消息队列中
        bool insert(const MessageProperties *mp, const std::string &body, DeliveryMode mode)
        {
            std::unique_lock<std::mutex> ul(_mtx);

            // 构建消息结构体
            MessagePtr ptr = std::make_shared<Message>();
            if (mp == nullptr)
            {
                ptr->mutable_payload()->mutable_property()->set_id(NS_Helper::Uuid::uuid());
                ptr->mutable_payload()->mutable_property()->set_mode(mode);
                ptr->mutable_payload()->mutable_property()->set_routing_key("");
            }
            else
            {
                // DLOG("是否持久化：%d", mp->mode() == DeliveryMode::DURABLE);
                ptr->mutable_payload()->mutable_property()->set_id(mp->id());
                if (mode == DeliveryMode::UNDURABLE)
                    ptr->mutable_payload()->mutable_property()->set_mode(mode);
                else
                    ptr->mutable_payload()->mutable_property()->set_mode(mp->mode());
                ptr->mutable_payload()->mutable_property()->set_routing_key(mp->routing_key());
            }
            ptr->mutable_payload()->set_body(body);
            ptr->mutable_payload()->set_valid("1");
            // 判断消息是否要被持久化
            // DLOG("是否持久化：%d", ptr->payload().property().mode() == DeliveryMode::DURABLE);
            if (ptr->payload().property().mode() == DeliveryMode::DURABLE)
            {
                // 将消息持久化到文件中
                // DLOG("持久化消息到文件中：%s", body.c_str());
                if (!_persist.insert(ptr))
                {
                    ELOG("持久化消息失败，%s", ptr->payload().body().c_str());
                    return false;
                }
                // 同时添加到持久化消息哈希表中
                // ELOG("持久化消息成功");
                _durable_msgs[ptr->payload().property().id()] = ptr;
                _valid_count++;
                _total_count++;
            }
            // 将消息添加到待推送队列中
            _to_push_msgs.push_back(ptr);
            // ELOG("插入消息成功");

            return true;
        }

        // 消息被应答，从待确认消息哈希表中删除
        bool erase(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            if (_waitack_msgs.find(msg_id) == _waitack_msgs.end())
            {
                // 意味着待确认消息哈希表中没有该消息直接返回
                return true;
            }

            // 判断消息是否是持久化消息
            if (_waitack_msgs[msg_id]->payload().property().mode() == DeliveryMode::DURABLE)
            {
                // 从消息文件中删除该消息
                _persist.erase(_waitack_msgs[msg_id]);
                // 从持久化哈希表中删去该消息
                _durable_msgs.erase(msg_id);
                _valid_count--;

                // 每次删除完消息之后判断是否需要进行垃圾回收
                gc();
            }
            // 从待确认消息哈希表中删除该消息
            _waitack_msgs.erase(msg_id);

            return true;
        }
        // 当服务器发送消息给订阅客户端时，就会使用这个接口，所以这个接口不仅仅是获取队首消息，也是弹出队首消息，即发送消息
        // 推送消息，消息从待推送消息队列转移到待确认消息队列中
        MessagePtr front()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            if (_to_push_msgs.size() == 0)
                return std::make_shared<Message>();

            MessagePtr ptr = _to_push_msgs.front();
            _to_push_msgs.pop_front();

            _waitack_msgs[ptr->payload().property().id()] = ptr;
            return ptr;
        }

        size_t getToPushMsgCount()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _to_push_msgs.size();
        }
        size_t getWaitACKMsgCount()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _waitack_msgs.size();
        }
        size_t getDurableMsgCount()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _durable_msgs.size();
        }
        size_t getTotalMsgCount()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _waitack_msgs.size() + _to_push_msgs.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            _persist.deleteFile();
            _to_push_msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }

        ~QueueMessage() {}

    private:
        void gc()
        {
            // DLOG("队列消息持久化消息总数量：%d, 队列消息持久化有效消息数量：%d", _total_count, _valid_count);
            if (_total_count < 2000 || _valid_count * 10 / _total_count > 5)
                return;

            // std::cout << "回收垃圾消息" << std::endl;
            // DLOG("回收垃圾消息, %d", _valid_count * 10 / _total_count);

            // 获取持久化队列消息文件中有效的消息
            std::list<MessagePtr> msgs = _persist.gc();
            for (auto &ptr : msgs)
            {
                std::string id = ptr->payload().property().id();
                if (_durable_msgs.find(id) == _durable_msgs.end())
                {
                    ELOG("队列消息文件中出现持久化哈希表中不存在的消息，%s", ptr->payload().body().c_str());
                    continue;
                }

                _durable_msgs[id]->set_offset(ptr->offset());
                _durable_msgs[id]->set_len(ptr->len());
            }
            _total_count = _valid_count = msgs.size();
        }

    private:
        std::mutex _mtx;
        std::string _qname;
        MessagePersist _persist;
        // 待推送消息列表
        std::list<MessagePtr> _to_push_msgs;
        // 持久化消息哈希表(方便当删除的消息是持久化消息时，能够找到快速找到文件中的位置并置为无效)
        std::unordered_map<std::string, MessagePtr> _durable_msgs;
        // 待确认消息哈希表
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;
        // 持久化有效消息数量
        int _valid_count;
        // 持久化消息总数量
        int _total_count;
    };

    class QueueMessageManager
    {
    public:
        using Ptr = std::shared_ptr<QueueMessageManager>;

        QueueMessageManager(const std::string &basedir) : _basedir(basedir) {}
        bool initQueueMessage(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 创建QueueMessage对象
                // 如果队列已经存在则无需初始化
                if (_queue_msgs.find(qname) != _queue_msgs.end())
                    return true;
                ptr = std::make_shared<QueueMessage>(_basedir, qname);
                // 将对象添加到队列单元管理哈希表中
                _queue_msgs[qname] = ptr;
            }
            // 队列消息的持久化数据恢复
            ptr->recovery();
            return true;
        }
        bool destoryQueueMessage(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                    return true;

                ptr = _queue_msgs[qname];
                _queue_msgs.erase(qname);
            }

            ptr->clear();

            return true;
        }
        bool insert(const std::string &qname, MessageProperties *mp, const std::string &body, DeliveryMode mode)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，无法插入消息，%s", qname.c_str());
                    return false;
                }
                ptr = _queue_msgs[qname];
            }
            return ptr->insert(mp, body, mode);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，无法获取消息，%s", qname.c_str());
                    return MessagePtr();
                }
                ptr = _queue_msgs[qname];
            }

            return ptr->front();
        }
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，无法确认消息，%s", qname.c_str());
                    return;
                }
                ptr = _queue_msgs[qname];
            }

            ptr->erase(msg_id);
        }
        size_t getToPushMsgCount(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，获取待推送消息数量，%s", qname.c_str());
                    return 0;
                }
                ptr = _queue_msgs[qname];
            }

            return ptr->getToPushMsgCount();
        }
        size_t getWaitACKMsgCount(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，获取待确认消息数量，%s", qname.c_str());
                    return 0;
                }
                ptr = _queue_msgs[qname];
            }

            return ptr->getWaitACKMsgCount();
        }
        size_t getDurableMsgCount(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，获取队列单元持久化消息数量，%s", qname.c_str());
                    return 0;
                }
                ptr = _queue_msgs[qname];
            }

            return ptr->getDurableMsgCount();
        }
        size_t getTotalMsgCount(const std::string &qname)
        {
            QueueMessage::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                // 如果队列不存在说明出现异常，直接返回
                if (_queue_msgs.find(qname) == _queue_msgs.end())
                {
                    ELOG("目标队列消息单元不存在，获取队列单元全部消息数量，%s", qname.c_str());
                    return 0;
                }
                ptr = _queue_msgs[qname];
            }

            return ptr->getTotalMsgCount();
        }

        size_t size()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _queue_msgs.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            for (auto &[qname, queue_masgs] : _queue_msgs)
            {
                queue_masgs->clear();
            }
            _queue_msgs.clear();
        }

        ~QueueMessageManager() {}

    private:
        std::mutex _mtx;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::Ptr> _queue_msgs;
    };

}

#endif